package com.arch.flume.interceptor;

import com.alibaba.fastjson2.JSON;
import com.google.common.base.Preconditions;
import org.apache.commons.lang3.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @author pizhihui
 * @date 2023-08-20
 */
public class DolphinSchedulerPathInterceptor implements Interceptor {
    private static final Logger LOG = LoggerFactory.getLogger(DolphinSchedulerPathInterceptor.class);

    // public static final String REGEX = "^.*(?<logpath>\\d{4}\\d{2}\\d{2}/.*)$";
    public static final String LOG_PATH_REGEX = "logPathRegex";
    public static final String LOG_PATH_BASE = "logPathBase";

    private Pattern regex;
    private String logPathBase;

    public DolphinSchedulerPathInterceptor(Pattern regex, String logPathBase) {
        this.regex = regex;
        this.logPathBase = logPathBase;
    }

    @Override
    public void initialize() {
        // NO-OP...
    }

    @Override
    public Event intercept(Event event) {

        Map<String, String> headers = event.getHeaders();
        String file = headers.get("file");
        if (StringUtils.isBlank(file)) {
            return event;
        }
        // /home/product/logs/dolphinscheduler_worker_logs/20230721/1400000000104849_3/5889251/8683561.log
        // /home/product/logs/dolphinscheduler_worker_logs/20230721/2156874943283200_1/5884451/8678005.log

        Matcher matcher = regex.matcher(file);
        if (matcher.find()) {
            String logpath = matcher.group("logpath");
            String newLogPath = logpath.replace(".", "_");
            headers.put("hdfsPath", newLogPath);
        } else {
            String newPath = file.replace(logPathBase + "/", "");
            String[] s = newPath.split("_");
            headers.put("hdfsPath", "out/" + s[0]);
        }
        // LOG.info("event.headers..... {}", JSON.toJSONString(event.getHeaders()));
        return event;
    }


    @Override
    public List<Event> intercept(List<Event> list) {
        return list.stream().map(this::intercept)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    @Override
    public void close() {
        // NO-OP...
    }

    public static class Builder implements Interceptor.Builder {

        Pattern regex;
        String logPathBase;
        @Override
        public Interceptor build() {
            return new DolphinSchedulerPathInterceptor(regex, logPathBase);
        }

        @Override
        public void configure(Context context) {
            String delim = context.getString(LOG_PATH_REGEX);
             logPathBase = context.getString(LOG_PATH_BASE);
            Preconditions.checkArgument(!org.apache.commons.lang.StringUtils.isEmpty(delim), "Must supply a valid delimiter");
            regex = Pattern.compile(delim);


        }
    }

}
